/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package marsdonne;

import marsdonne.tasks.SqlTask;
import org.apache.flink.api.java.ExecutionEnvironment;

import java.sql.Types;

/**
 * Skeleton for a Flink Batch Job.
 *
 * <p>For a tutorial how to write a Flink batch application, check the
 * tutorials and examples on the <a href="http://flink.apache.org/docs/stable/">Flink Website</a>.
 *
 * <p>To package your application into a JAR file for execution,
 * change the main class in the POM.xml file to this class (simply search for 'mainClass')
 * and run 'mvn clean package' on the command line.
 */
public class BatchJob {

    public static final String sql = "insert into metrics_monthly(type, timed, metadata, contents)" +
            "(select '机构签约率' as type, timestamptz(?) as timed, '{\"ver\":1.0,\"org\":\"基层机构\",\"r_num\":\"居民数\",\"s_num\":\"签约数\",\"a_num\":\"激活数\"}'::jsonb as metadata, array_to_json(array_agg(t)) as contents from" +
            "(SELECT" +
            " ol.contents -> 'name' AS org," +
            " COUNT ( rl.contents -> 'id' ) AS r_num ," +
            " COUNT ( rms.contents -> 'resident_id' ) AS s_num," +
            " SUM ( CASE rl.contents -> 'account_status' WHEN '2' THEN 1 ELSE 0 END ) AS a_num " +
            "FROM" +
            " ( SELECT contents FROM data_business WHERE TYPE = 'resident_manage_status' and timed BETWEEN timestamptz(?) and timestamptz(?)) rms ," +
            " ( SELECT contents FROM data_standard WHERE TYPE = 'resident_list') rl," +
            " (" +
            " SELECT  contents  FROM  data_standard  WHERE TYPE = 'org_list'  AND contents -> 'kind' = '\"H\"' AND ( contents -> 'name' = '\"铜井社区卫生服务中心\"' OR contents -> 'name' = '\"横溪街道社区卫生服务中心\"' )) ol " +
            "WHERE ol.contents -> 'id' = rl.contents -> 'org_id' AND rms.contents -> 'resident_id' = rl.contents -> 'id' GROUP BY org) t)" +
            "on conflict(type,timed)" +
            "do UPDATE set contents = metrics_monthly.contents;";

    public static void main(String[] args) throws Exception {
        new SqlTask(SqlTask.DbType.PgSql, sql, new int[]{Types.VARCHAR, Types.VARCHAR, Types.VARCHAR},"2018-05-01,2018-05-31|2018-10-01,2018-10-31|2018-12-01,2018-12-31").submitTo(ExecutionEnvironment.getExecutionEnvironment());
    }
}
